package xubo.wangcaifeng.love


import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Duration, StreamingContext}
import scalikejdbc.{DB, SQL}
import scalikejdbc.config.DBs



object Custormer {
  def main(args: Array[String]): Unit = {
    //首先我们得先从数据库中读取偏移量，如果有偏移量，就拿到偏移量，然后去kafka中读取数据
    //指定组名
    val group = "xbwcf"
    //创建SparkConf
    val conf = new SparkConf().setAppName("MoveDate").setMaster("local[*]")
    //创建SparkStreaming，并设置间隔时间
    val ssc = new StreamingContext(conf,Duration(5000))
    //指定消费的topic
    val topic = "movedata"
    //指定kafka的broker地址
    val brokerList = "mini1:9092,mini2:9092,mini3:9092"
    //创建stream时使用的topic名字集合
    val topics: Set[String] = Set(topic)
    //准备kafka的参数
    val kafkaParams = Map(
      "metadata.broker.list" -> brokerList,
      "group.id" -> group, //指定消费者组的ID
      "auto.offset.reset" -> kafka.api.OffsetRequest.SmallestTimeString //从最开始进行消费
    )
    //获取数据库中的偏移量，如果有，则对应的拿kafka中的数据，如果没有就从头消费
    //连接数据库  判断分区总和是否为零
    DBs.setup()
    val partition: List[Int] = DB.readOnly(
      implicit session =>
        SQL(s"select count(partition) count from move where topic = '${topic}' and groupid = '${group}'").map(rs => rs.int("count")).list().apply()
    )
    //判断partition的长度，如果partition的长度为零，则根据kafkaparam的配置进行读取数据
    var kafkaStream: InputDStream[(String, String)] = null
    //如果数据库中有偏移量，我们会利用这个偏移量作为 kafkaStream 的起始位置
    var fromOfferset: Map[TopicAndPartition,Long] = Map()
    //如果保存过 offset，这里更好的做法，还应该和  kafka 上最小的 offset 做对比，不然会报 OutOfRange 的错误
    if (partition(0)>0){
      //如果partition的长度大于零，则曾经读取果偏移量，从数据库中查到对应分区的偏移量，进行数据消费
      for (i <- 0 until partition.length){
          //取偏移量
        val offerset: List[Int] = DB readOnly ({
          implicit sesseion =>
            SQL(s"select offerset from move where topic = '${topic}' and groupid = '${group}' and partition = '${partition(i)}'").map(res => res.int("offerset")).list().apply()
        })
        val tp = TopicAndPartition(topic,i)
        //将不同partition的对应的偏移量更新到fromOfferset中
        fromOfferset += (tp -> offerset(i).toLong)
      }
      //现在要将数据数据继续消费，将kafka中的数据变成（topic_name,message）
      val messageHandler = (mmd:MessageAndMetadata[String,String]) => (mmd.topic,mmd.message())
      //通过kafkaUtils创建直连的DStream，继续消费数据
      KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder,(String,String)](ssc,kafkaParams,fromOfferset,messageHandler)
    } else{
      kafkaStream = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)
    }
    //kafaka直连DStream，依次拿出Rdd
    kafkaStream.foreachRDD(
      rdd =>{
        //本质上这个RDD是kafkaRDD,将其转换成HasOffsetRange,然后获取偏移量
        //RDD有多个分区，每一个分区都对应一个偏移量,获取到每个分区的偏移量
        val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //然后对该RDD进行处理，对某个批次的RDD进行处理
        val lines: RDD[String] = rdd.map(_._2)
        //RDD里面有很多的数据，对其进行处理
        //统计订单量
        Utils.orderSum(lines)
        //计算充值金额
        Utils.SumMoney(lines)
        //计算充值成功率
        Utils.SuccessRate(lines)
        //计算平均时长
        Utils.avgTime(lines)
        //计算全部的订单量
        Utils.topOrder(lines)
        //计算全部的成功率
        Utils.topOrderSuccess(lines)
        //计算各个省的失败量
        Utils2.EachProviceFailCount(lines)
        //计算每分钟的订单量和充值金额
        Utils2.EachSecondMoney(lines)
        Utils2.EachSecondOrder(lines)
        //将偏移量写入到mysql中
        for (o <- offsetRanges) {
          //更新各分区的偏移量
          DB.autoCommit{implicit session =>
            SQL("replace into move(groupid,topic,partition,offerset) values(?,?,?,?)").bind(topic,group,o.partition,o.untilOffset).update().apply()
          }
        }
      })
    ssc.start()
    ssc.awaitTermination()



    //当消费完数据，把偏移量记录到数据库中，下一次从这消费
  }

}
